package com.qyer.lxp.common.framework.threadpool;

import org.apache.log4j.Logger;

import java.io.Serializable;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 扩展ThreadPoolExecutor<br/>
 * 维护一个线程池中未完成任务的计数器
 *
 * @author jiangping
 */
public class RunningCounterThreadPoolExecutor
        extends ThreadPoolExecutor implements Serializable {

    private final Logger logger = Logger.getLogger(this.getClass());
    /**
     * 未完成的任务数
     */
    final AtomicInteger uncompletedTasksCount = new AtomicInteger();

    RunningCounterThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
                                     long keepAliveTime, TimeUnit unit,
                                     BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory,
                                     RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                threadFactory, handler);

    }

    public AtomicInteger getUncompletedTasksCount() {
        return this.uncompletedTasksCount;
    }

    @Override
    public void execute(Runnable command) {
        //执行前任务计数器加1
        uncompletedTasksCount.incrementAndGet();
        super.execute(command);
    }

    @Override
    protected <T> RunnableFuture<T> newTaskFor(
            Callable<T> callable) {
        if (callable instanceof IAroundAdviceCallable) {
            return new AroundAdviceFutureTask<T>(callable);
        } else {
            return super.newTaskFor(callable);
        }
    }

    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        //执行完毕任务计数器减1
        uncompletedTasksCount.decrementAndGet();
        //修复针对任务执行异常但异常信息丢失的问题
        Future<?> f = (Future<?>) r;
        try {
            f.get();
        } catch (Exception e) {
            t = e;
            logger.error("任务（" + r + "）执行异常：" + t);
        }
        if (r instanceof AroundAdviceFutureTask) {
            IAroundAdviceCallable aroundAdviceCallable =
                    ((AroundAdviceFutureTask) r).getArounAdviceCallable();
            aroundAdviceCallable.executeAfter(t);
        }
    }

    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        if (r instanceof AroundAdviceFutureTask) {
            IAroundAdviceCallable aroundAdviceCallable =
                    ((AroundAdviceFutureTask) r).getArounAdviceCallable();
            aroundAdviceCallable.executeBefore(t);
        }
    }
}
